package com.blyat.xsoft.kernel.component.disruptor;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.Objects;
import java.util.concurrent.ThreadFactory;

/**
 * @Auther: syh
 * @Date: 2020/7/8
 * @Description: 高性能异步处理框架
 */
public class MetadataPublisher<T> implements Publisher<T> {

    private RingBuffer<Metadata<T>> ringBuffer;
    private Disruptor<Metadata<T>> disruptor;
    private EventHandlerGroup<Metadata<T>> handlerGroup;
    private PublisherState state;
    private PublishStrategy publishStrategy;
    private MetadataTranslator<T> translator;
    private Producer<T>[] producers;

    private enum PublisherState {
        START, SHUTDOWN
    }

    public MetadataPublisher(PublisherBuilder<T> builder) {
        this(builder.getMetadataFactory(),
                builder.getRingSize(),
                builder.getThreadFactory(),
                builder.getWaitStrategy(),
                builder.getErrorHandler(),
                builder.getType(),
                builder.getPublishStrategy());
    }

    public MetadataPublisher(MetadataFactory<T> factory,
                             int ringSize,
                             ThreadFactory threadFactory,
                             WaitStrategy waitStrategy,
                             ErrorHandler<T> errorHandler,
                             ProducerType type,
                             PublishStrategy publishStrategy) {

        this.disruptor = new Disruptor<>(factory, ringSize, threadFactory,
                type, waitStrategy);

        if (!Objects.isNull(errorHandler)) {
            this.disruptor.setDefaultExceptionHandler(errorHandler);
        }

        this.ringBuffer = disruptor.getRingBuffer();

        this.publishStrategy = publishStrategy;
        if (publishStrategy == PublishStrategy.TRANSLATOR) {
            translator = new MetadataTranslator<>();
        }

        this.state = PublisherState.SHUTDOWN;
    }

    public MetadataPublisher<T> producer(MetadataProducer<T> ... producers) {
        if (!Objects.isNull(producers) && producers.length > 0) {
            for (MetadataProducer<T> producer : producers) {
                producer.setRingBuffer(this.ringBuffer).setTranslator(this.translator);
            }
            this.producers = producers;
        }
        return this;
    }

    public MetadataPublisher<T> eventHandler(EventHandler<Metadata<T>>... eventHandlers) {
        if (this.handlerGroup != null) {
            this.handlerGroup.handleEventsWith(eventHandlers);
        } else {
            this.handlerGroup = disruptor.handleEventsWith(eventHandlers);
        }
        return this;
    }

    public MetadataPublisher<T> workHandler(WorkHandler<Metadata<T>>... workHandlers) {
        if (this.handlerGroup != null) {
            this.handlerGroup.handleEventsWithWorkerPool(workHandlers);
        } else {
            this.handlerGroup = disruptor.handleEventsWithWorkerPool(workHandlers);
        }
        return this;
    }

    public MetadataPublisher<T> thenHandler(EventHandler<Metadata<T>>... thenHandlers) {
        this.handlerGroup.then(thenHandlers);
        return this;
    }

    public MetadataPublisher<T> afterHandler(EventHandler<Metadata<T>>... afterHandlers) {
        this.handlerGroup = this.disruptor.after(afterHandlers);
        return this;
    }

    public Disruptor<Metadata<T>> getDisruptor() {
        return disruptor;
    }

    @Override
    public MetadataPublisher<T> start() {
        this.disruptor.start();
        this.state = PublisherState.START;
        return this;
    }

    @Override
    public MetadataPublisher<T> shutdown() {
        this.disruptor.shutdown();
        this.state = PublisherState.SHUTDOWN;
        return this;
    }

    @Override
    public MetadataPublisher<T> publish(T t) {
        if (!isStarted()) {
            throw new IllegalStateException("publisher not start..");
        }
        if (producers == null || producers.length <= 0) {
            throw new IllegalStateException("producer must be specify.");
        }

        for (Producer<T> producer : producers) {
            producer.produce(t);
        }
        return this;
    }

    private boolean isStarted() {
        return this.state == PublisherState.START;
    }
}
